package com.jml.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.junit.Test;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;


public class HdfsClient{
    //-DHADOOP_USER_NAME=root  服务器配置的用户名

    /**
     * 文件上传
     */
    @Test
    public void testCopyFromLocalFile() throws IOException, InterruptedException, URISyntaxException {
        for (int i = 0; i < 1; i++) {
            // 1 获取文件系统
            //就是core-site.xml中的Configuration标签
            Configuration configuration = new Configuration();
            //设置副本个数
            configuration.set("dfs.replication", "2");
            //设置用户
            FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:8020"), configuration, "root");
            // 2 上传文件，把本地的D:/banzhang.txt文件上传到HDFS的根目录，命名为banzhang.txt
            fs.copyFromLocalFile(new Path("D:/banzhang.txt"), new Path("/banzhang"+i+".txt"));
            // 3 关闭资源
            fs.close();
            System.out.println("over");
        }
    }

    /**
     * 文件下载
     */
    @Test
    public void testCopyToLocalFile() throws IOException, InterruptedException, URISyntaxException{
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:8020"), configuration, "root");
        // 2 执行下载操作
        // boolean delSrc 指是否将原文件删除
        // Path src 指要下载的文件路径
        // Path dst 指将文件下载到的路径
        // boolean useRawLocalFileSystem 是否开启文件校验
        fs.copyToLocalFile(false, new Path("/banzhang.txt"), new Path("D:/banhuajml.txt"), true);
        // 3 关闭资源
        fs.close();
    }

    /**
     * 创建目录mkdir
     */
    @Test
    public void testMkdirs() throws IOException, InterruptedException, URISyntaxException {
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        // 配置在集群上运行
        // configuration.set("fs.defaultFS", "hdfs://hadoop101:8020");
        // FileSystem fs = FileSystem.get(configuration);
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:8020"), configuration, "root");
        // 2 创建目录
        fs.mkdirs(new Path("/1108/daxian/banzhang"));
        // 3 关闭资源
        fs.close();
    }

    /**
     * 文件夹删除
     */
    @Test
    public void testDelete() throws IOException, InterruptedException, URISyntaxException{
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:8020"), configuration, "root");
        // 2 执行删除
        fs.delete(new Path("/1108/"), true);
        // 3 关闭资源
        fs.close();
    }

    /**
     * HDFS文件名更改
     */
    @Test
    public void testRename() throws IOException, InterruptedException, URISyntaxException{
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:8020"), configuration, "root");
        // 2 修改文件名称
        fs.rename(new Path("/banzhang.txt"), new Path("/banhua.txt"));
        // 3 关闭资源
        fs.close();
    }

    /**
     * HDFS文件详情查看
     */
    @Test
    public void testListFiles() throws IOException, InterruptedException, URISyntaxException{
        // 1获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:8020"), configuration, "root");
        // 2 获取文件详情，得到所有的文件
        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);
        while(listFiles.hasNext()){
            LocatedFileStatus status = listFiles.next();
            // 输出详情
            // 文件名称
            System.out.println("Name:"+status.getPath().getName());
            // 长度
            System.out.println("Len:"+status.getLen());
            // 权限
            System.out.println("Permission:"+status.getPermission());
            // 分组
            System.out.println("Group:"+status.getGroup());
            // 获取存储的块信息
            BlockLocation[] blockLocations = status.getBlockLocations();
            for (BlockLocation blockLocation : blockLocations) {
                // 获取块存储的主机节点
                String[] hosts = blockLocation.getHosts();
                for (String host : hosts) {
                    System.out.println("blockHost:"+host);
                }
            }
            System.out.println("-----------文件分割线----------");
        }
        // 3 关闭资源
        fs.close();
    }

    /**
     * HDFS文件和文件夹判断
     */
    @Test
    public void testListStatus() throws IOException, InterruptedException, URISyntaxException{
        // 1 获取文件配置信息
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:8020"), configuration, "root");
        // 2 判断是文件还是文件夹,获取所有文件的状态
        FileStatus[] listStatus = fs.listStatus(new Path("/"));
        for (FileStatus fileStatus : listStatus) {
            // 如果是文件
            if (fileStatus.isFile()) {
                System.out.println("f:"+fileStatus.getPath().getName());
            }else {
                System.out.println("d:"+fileStatus.getPath().getName());
            }
        }
        // 3 关闭资源
        fs.close();
    }

    /**
     * 追加
     */
    @Test
    public void testAppend() throws IOException, InterruptedException, URISyntaxException{
        // 1 获取文件配置信息
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:8020"), configuration, "root");
        FSDataOutputStream append = fs.append(new Path("/README.txt"));
        append.write("TextAPI".getBytes());
        IOUtils.closeStream(append);
        // 3 关闭资源
        fs.close();
    }

}
